// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/dwio/common/selective_column_reader_internal.h>
#include <pollux/dwio/dwrf/reader/dwrf_data.h>

namespace kumo::pollux::dwrf {

class SelectiveStringDirectColumnReader
    : public dwio::common::SelectiveColumnReader {
 public:
  using ValueType = StringView;
  SelectiveStringDirectColumnReader(
      const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
      DwrfParams& params,
      common::ScanSpec& scanSpec);

  void seekToRowGroup(int64_t index) override {
    SelectiveColumnReader::seekToRowGroup(index);
    auto positionsProvider = formatData_->seekToRowGroup(index);
    blobStream_->seekToPosition(positionsProvider);
    lengthDecoder_->seekToRowGroup(positionsProvider);

    POLLUX_CHECK(!positionsProvider.hasNext());

    bytesToSkip_ = 0;
    bufferStart_ = bufferEnd_;
  }

  uint64_t skip(uint64_t numValues) override;

  void read(int64_t offset, const RowSet& rows, const uint64_t* incomingNulls)
      override;

  void getValues(const RowSet& rows, VectorPtr* result) override {
    rawStringBuffer_ = nullptr;
    rawStringSize_ = 0;
    rawStringUsed_ = 0;
    getFlatValues<StringView, StringView>(rows, result, requestedType());
  }

 private:
  template <bool hasNulls>
  void skipInDecode(int32_t numValues, int32_t current, const uint64_t* nulls);

  melon::StringPiece readValue(int32_t length);

  template <bool hasNulls, typename Visitor>
  void decode(const uint64_t* nulls, Visitor visitor);

  template <typename TVisitor>
  void readWithVisitor(RowSet rows, TVisitor visitor);

  void extractCrossBuffers(
      const int32_t* lengths,
      const int64_t* starts,
      int32_t rowIndex,
      int32_t numValues);

  inline void makeSparseStarts(
      int32_t startRow,
      const int32_t* rows,
      int32_t numRows,
      int64_t* starts);

  inline void extractNSparse(const int32_t* rows, int32_t row, int numRows);

  void extractSparse(const int32_t* rows, int32_t numRows);

  template <bool scatter, bool skip>
  bool try8Consecutive(int64_t start, const int32_t* rows, int32_t row);

  template <bool kScatter, bool kGreaterThan4>
  bool
  try8ConsecutiveSmall(const char* data, const uint16_t* offsets, int startRow);

  std::unique_ptr<dwio::common::IntDecoder</*isSigned*/ false>> lengthDecoder_;
  std::unique_ptr<dwio::common::SeekableInputStream> blobStream_;
  const char* bufferStart_ = nullptr;
  const char* bufferEnd_ = nullptr;
  BufferPtr lengths_;
  int32_t lengthIndex_ = 0;
  const uint32_t* rawLengths_ = nullptr;
  int64_t bytesToSkip_ = 0;
  // Storage for a string straddling a buffer boundary. Needed for calling
  // the filter.
  std::string tempString_;
};

} // namespace kumo::pollux::dwrf
